package org.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.atguigu.gmall.realtime.bean.TableProcess;
import org.atguigu.gmall.realtime.util.JdbcUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.sink
 * Author: ZengHaiFeng
 * CreateTime:2023/4/22 11:13:29
 * Description: Todo build phoenix sink
 * Version：1.0
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    private Connection connection;
    @Override
    public void open(Configuration parameters) {
        connection = JdbcUtil.getPhoenixConnection();
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        JSONObject data = value.f0;
        TableProcess conf = value.f1;

        if (data.getString("op_type").equals("delete")){
            delete(data,conf);
        }else {
            upsert(data,conf);
        }
    }

    /**
     * @description: delete phoenix row that op_type is delete
     * @author: zenghaifeng
     * @date: 2023/4/22 11:37
     * @param data
     * @param conf
     **/
    private void delete(JSONObject data, TableProcess conf) throws SQLException {
        // build delete sql : delete from tableName where id = ?
        StringBuilder sql = new StringBuilder();
        sql
                .append("delete from ")
                .append(conf.getSinkTable())
                .append(" where id = ? ");

        // prepareStatement
        PreparedStatement pre = connection.prepareStatement(sql.toString());
        // set props(?)
        pre.setString(1, data.getString("id"));
        // commit transaction
        connection.commit();
        pre.close();
    }

    /**
     * @description: insert or update phoenix row
     * @author: zenghaifeng
     * @date: 2023/4/22 11:39
     * @param data
     * @param conf
     **/
    private void upsert(JSONObject data, TableProcess conf) throws SQLException {
        // upsert into tableName (columns) values(data)
        StringBuilder sql = new StringBuilder();
        sql
                .append("upsert into ")
                .append(conf.getSinkTable())
                .append(" (")
                .append(conf.getSinkColumns())
                .append(" ) values( ")
                .append(conf.getSinkColumns().replaceAll("[^,]+","?"))
                .append(" ) ");
        PreparedStatement pre = connection.prepareStatement(sql.toString());
        String[] columns = conf.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            pre.setString(i+1, data.getString(columns[i]));
        }
        pre.execute();
        connection.commit();
        pre.close();
    }
}
